package com.example.demo.rocketmq_client;

import com.alibaba.fastjson.JSON;
import com.example.demo.VO.User;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

public class TransactionMQProducerDemo {
    public static void main(String[] args) throws Exception {

        //1.创建消息生产者producer，并制定生产者组名
        TransactionMQProducer producer = new TransactionMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("127.0.0.1:9876");

        //添加事务监听器
        producer.setTransactionListener(new TransactionListener() {
            /**
             * 在该方法中执行本地事务
             * @param msg
             * @param arg
             * @return
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {

                if ("taga".equals(msg.getKeys())) {
                    System.out.println("===========本地事务 taga=============");
                    try {
                        Thread.sleep(10000);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }

                    return LocalTransactionState.COMMIT_MESSAGE;
                } else if ("tagb".equals(msg.getKeys())) {
                    System.out.println("===========本地事务 tagb=============");
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } else if ("tagc".equals(msg.getKeys())) {
                    System.out.println("===========本地事务 tagc=============");
                    return LocalTransactionState.UNKNOW;
                }
                return LocalTransactionState.UNKNOW;
            }

            /**
             * 该方法时MQ进行消息事务状态回查
             * @param msg
             * @return
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("===========事务回查=============");
                System.out.println("消息的Tag:" + msg.getTags());
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        //3.启动producer
        producer.start();


        //4.创建消息对象，指定主题Topic、Tag和消息体
        /**
         * 参数一：消息主题Topic
         * 参数二：消息Tag
         * 参数三：消息内容
         */
        User user = new User();
        user.setId("123456");
        user.setName("消息");
        Message msg = new Message("Topic1", "Tag1", JSON.toJSONString(user).getBytes());
        msg.setKeys("taga");
        //5.发送消息
        SendResult result = producer.sendMessageInTransaction(msg, null);
        //发送状态
        SendStatus status = result.getSendStatus();
        System.out.println("发送结果:" + JSON.toJSONString(status));

        MessageQueue m = new MessageQueue();
        m.getTopic();
        m.getBrokerName();
        m.getQueueId();

        //6.关闭生产者producer
        //producer.shutdown();
    }

}
